package net.bwie.realtime.jtp.ods.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;

public class CdcDataEtlMapFunction extends RichMapFunction<String, String> {
    @Override
    public String map(String value) throws Exception {
        /*
        {
            "before": null,
            "after": Object{...},
            "source": {
                "version": "1.9.7.Final",
                "connector": "mysql",
                "name": "mysql_binlog_source",
                "ts_ms": 0,
                "snapshot": "false",
                "db": "jtp_mall",
                "sequence": null,
                "table": "user_info",
                "server_id": 0,
                "gtid": null,
                "file": "",
                "pos": 0,
                "row": 0,
                "thread": null,
                "query": null
            },
            "op": "r",
            "ts_ms": 1748843956799,
            "transaction": null
        }
         */
        JSONObject jsonObject = JSON.parseObject(value);

        JSONObject before = jsonObject.getJSONObject("before");
        JSONObject after = jsonObject.getJSONObject("after");
        String op = jsonObject.getString("op");
        Long ts_ms = jsonObject.getLong("ts_ms");
        JSONObject source = jsonObject.getJSONObject("source");
        String db = source.getString("db");
        String table = source.getString("table");

        String operateType;
        switch (op) {
            case "u":
                operateType = "update";
                break;
            case "r":
            case "c":
                operateType = "insert";
                break;
            case "d":
                operateType = "delete";
                break;
            default:
                operateType = "unknown";
        }

        JSONObject newJsonObject = new JSONObject();
        newJsonObject.put("operate_type", operateType);
        newJsonObject.put("ts_ms", ts_ms);
        newJsonObject.put("db_name", db);
        newJsonObject.put("table_name", table);

        if ("d".equals(op)) {
            newJsonObject.put("operate_data", before);
        } else {
            newJsonObject.put("operate_data", after);
        }

        return newJsonObject.toJSONString();
    }
}
